package com.hnyd.streaming

import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.{lang, util}
import scala.collection.JavaConverters.asJavaIterableConverter
import scala.util.matching.Regex

object mooc_real_app {
  var total = 0
  var tprice: Float = 0.0F
  var ttime = 0
  var tscore = 0
  var tnums = 0

  def main(args: Array[String]): Unit = {
    // 准备配置对象
    val sparkConf = new SparkConf().setAppName("mooc_real_app").setMaster("local[*]")
    // 构建上下文环境对象
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.sparkContext.setCheckpointDir("cp")
    val kafkaDream: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream("spider-topic", ssc)
    val moocKafkaMessage: DStream[MoocKafkaMessage] = kafkaDream.map {
      record => {
        val message: String = record.value()
        val datas: Array[String] = message.split(",")
        //处理数据，价格￥398.00 原价￥399.00
        val r: Regex = "\\d+(\\.\\d+)?".r
        val price: Option[String] = r.findFirstIn(datas(1))
        //处理时长
        var timer = datas(3)
        timer = timer.replace(" ", "")
        val i = timer.indexOf("分钟")
        var mi = 0
        if (-1 < i) {
          val timerArr = timer.replace("分钟", "").replace(" ", "").split("小时")
          mi = (timerArr(0).toInt * 60 * 60) + timerArr(1).toInt * 60
        } else {
          val timerArr = timer.replace("小时", "").toInt
          mi = timerArr * 60 * 60
        }

        MoocKafkaMessage(datas(0), price.get.toFloat, datas(2), mi, datas(4).toInt, datas(5).toInt)
      }
    }
    //对难易程度分组，统计总时长，总的学习人数，总评分数据，总课程数
    val stateDream: DStream[(String, (Int, Int, Int))] = moocKafkaMessage.
      map(x => (x.level, (x.timer, x.nums, x.score)))
      .updateStateByKey((seq: Seq[(Int, Int, Int)], option: Option[(Int, Int, Int)]) => {
        var n1 = 0
        var n2 = 0
        var n3 = 0
        val tup: (Int, Int, Int) = option.getOrElse((0, 0, 0))
        n1 = tup._1
        n2 = tup._2
        n3 = tup._3
        for (s <- seq) {
          n1 += s._1
          n2 += s._2
          n3 += s._3
        }
        Option(n1, n2, n3)
      })
    //转换map为JAVA类型
    import scala.collection.JavaConverters.mapAsJavaMapConverter
    stateDream.foreachRDD(x => x.foreachPartition(pair => pair
      .foreach(x => {

        var map: util.Map[String, String] = Map(
          "timer" -> x._2._1.toString,
          "nums" -> x._2._2.toString,
          "score" -> x._2._3.toString).asJava
        val inner: Jedis = RedisUtil.getJedisClient
        inner.hmset("level:" + x._1, map)
        inner.close()
      })))


    //统计学习人数前10的课程 每隔20s计算最近10s的课程
    // xindowDuration:窗口长度/窗口大小，表示要计算最近多长时间的数据
    // slideDuration:滑动间隔，表示每隔多长时间计算一次
    //    val resultDS: DStream[(String, Int)] = moocKafkaMessage.map(x => (x.title, x.nums)).reduceByKeyAndWindow((a:Int, b:Int) => a + b, Seconds(20), Seconds(10))
    //    //注意DStream没有提供直接排序的方法，所以需要直接对底层的RDD进行操作
    //    //DStream的transform方法表示对DStream底层的RDD进行操作并返回结果
    //    val sortRdd: DStream[(String, Int)] = resultDS.transform(rdd => {
    //      val sortRDD: RDD[(String, Int)] = rdd.sortBy(_._2, false)
    //      val top10: Array[(String, Int)] = sortRDD.take(10)
    //      println("======top10======")
    //      top10.foreach(println)
    //      sortRDD
    //    })


    //统计课程数，课程平均价格，总时长，总评分，总学习人数

        val totalDS: DStream[(Int, Float, Int, Int, Int)] = moocKafkaMessage.map(record => {
          total = total + 1
          tprice += record.price.toFloat
          ttime += record.timer
          tscore += record.score
          tnums += record.nums
          (total, tprice, ttime, tscore, tnums)
        })

        totalDS.foreachRDD(x=>x.foreachPartition(partition=>{
          partition.foreach(x=>{
            val map = Map("total" -> total.toString,"tprice" -> tprice.toString,"ttime"->ttime.toString,"tscore"->tscore.toString,"tnums"->tnums.toString)
            val inner: Jedis = RedisUtil.getJedisClient
            inner.hset("mooctotal","total",x._1.toString)
            inner.hset("mooctotal","price",x._2.toString)
            inner.hset("mooctotal","timer",x._3.toString)
            inner.hset("mooctotal","score",x._4.toString)
            inner.hset("mooctotal","nums",x._5.toString)
            inner.close()
          })
        }))


    // totalDS.print()
    ssc.start()
    ssc.awaitTermination()
  }

}
